package com.wavehi.producer;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.wavehi.util.RabbitMQFactoryUtils;

import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * @Description 生产者 异步确认发布
 * @Author WaveHi
 * @Date 26/12/2021 下午 9:04
 * @Version 1.0
 * @Contact HJBWaveHi@163.com
 */
public class Producer_03 {
    //队列名
    private static final String QUEUE_NAME = "release_confirmation";
    //消息数
    private static final int MESSAGE_COUNT = 1000;
    public static void main(String[] args) throws Exception {

        //1.获取连接信道
        Channel channel = RabbitMQFactoryUtils.getChannel();
        //2.声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        //3.开启发布确认
        channel.confirmSelect();
        ConcurrentSkipListMap<Long, String> concurrentSkipListMap = new ConcurrentSkipListMap<>();
        //4.准备发布成功的回调操作
        ConfirmCallback ackCallBack = (deliverTag, multiple) -> {
            System.out.println(concurrentSkipListMap.get(deliverTag) + " send completed!");
            if(multiple) {
                ConcurrentNavigableMap<Long, String> longStringConcurrentNavigableMap =
                        concurrentSkipListMap.headMap(deliverTag);
                longStringConcurrentNavigableMap.clear();
            }else {
                concurrentSkipListMap.remove(deliverTag);
            }
        };
        //5.准备发布失败的回调操作
        ConfirmCallback failCallBack = (deliverTag, multiple) -> {
            String s = concurrentSkipListMap.get(deliverTag);
            System.out.printf("Send message {} failed!", s);
        };
        //6.设置发布确认监听器
        channel.addConfirmListener(ackCallBack, failCallBack);
        //7.准备消息
        long begin = System.currentTimeMillis();
        String message = null;
        //8.发布消息
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            message = UUID.randomUUID().toString();
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            concurrentSkipListMap.put(channel.getNextPublishSeqNo(), message);
        }
        long end = System.currentTimeMillis();
        System.out.println("Time : " + (end - begin));//431
    }

}
